package com.intellif.mozping.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.support.Acknowledgment;

import java.util.Date;
import java.util.List;

/**
 * @author by mozping
 * @Classname LogRecord
 * @Description TODO
 * @Date 2019/4/16 19:12
 */
public class LogRecord {

    public static final Logger logger = LoggerFactory.getLogger(LogRecord.class);

    public static void logRecode(ConsumerRecord<?, ?> record) {
        if (record.offset() % 10000 == 0) {
            logger.info("消费数据: topic = {},partition = {}, offset = {},value = {},Time: " + new Date(),
                    record.topic(),
                    record.partition(),
                    record.offset(),
                    record.value().toString().substring(10, 20)
            );
        }
    }

    public static void handle(List<ConsumerRecord> consumerRecords, Acknowledgment ack) {
        try {
            for (ConsumerRecord<?, ?> record : consumerRecords) {
                LogRecord.logRecode(record);
            }
            ack.acknowledge();
        } catch (Exception e) {
            logger.error("归档统计异常：", e);
        }
    }
}
